#ifndef __MQ_CHANNEL_HPP__
#define __MQ_CHANNEL_HPP__

#include "mq_consumer.hpp"
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "../mq_comm/mq_proto.pb.h"
#include "../mq_comm/helper.hpp"
#include "../mq_comm/logger.hpp"
#include <iostream>
#include <mutex>
#include <condition_variable>

namespace mq
{
    using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;
    using MessagePtr=std::shared_ptr<google::protobuf::Message>;
    using basicConsumeResponsePtr=std::shared_ptr<basicConsumeResponse>;
    using basicCommonResponsePtr=std::shared_ptr<basicCommonResponse>;
    class Channel
    {
    public:
        using ptr=std::shared_ptr<Channel>;
        Channel(const muduo::net::TcpConnectionPtr& conn,const ProtobufCodecPtr& codec)
            :_cid(UUIDHelper::uuid())
            ,_conn(conn)
            ,_codec(codec)
            ,_consumer(nullptr)
        {}

        /*返回信道ID*/
        std::string cid()
        {
            return _cid;
        }

        /*打开信道*/
        bool openChannel()
        {
            std::string rid=UUIDHelper::uuid();
            openChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp=waitResponse(rid);
            return resp->ok();
        }

        /*关闭信道*/
        void closeChannel()
        {
            std::string rid=UUIDHelper::uuid();
            closeChannelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*声明交换机*/
        bool declareExchange(const std::string& name,
                            ExchangeType type,
                            bool durable,
                            bool auto_delete,
                            google::protobuf::Map<std::string,std::string>& args)
        {
            std::string rid=UUIDHelper::uuid();
            declareExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp=waitResponse(rid);
            return resp->ok();
        }

        /*删除交换机*/
        void deleteExchange(const std::string& name)
        {
            std::string rid=UUIDHelper::uuid();
            deleteExchangeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*声明队列*/
        bool declareQueue(const std::string& qname,
                        bool durable,
                        bool exclusive,
                        bool auto_delete,
                        google::protobuf::Map<std::string,std::string>& args)
        {
            std::string rid=UUIDHelper::uuid();
            declareQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp=waitResponse(rid);
            return resp->ok();
        }

        /*删除队列*/
        void deleteQueue(const std::string& qname)
        {
            std::string rid=UUIDHelper::uuid();
            deleteQueueRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(qname);
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*绑定队列*/
        bool queueBind(const std::string& ename,const std::string& qname,const std::string& binding_key)
        {
            std::string rid=UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(binding_key);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp=waitResponse(rid);
            return resp->ok();
        }

        /*解除绑定*/
        void queueUnBind(const std::string& ename,const std::string& qname)
        {
            std::string rid=UUIDHelper::uuid();
            queueBindRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*发布消息*/
        void basicPublish(const std::string& ename,const BasicProperties* bp,const std::string& body)
        {
            std::string rid=UUIDHelper::uuid();
            basicPublishRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_body(body);
            if(bp!=nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*消息确认,只有作为消费者才能够进行消息确认*/
        void basicAck(const std::string& msg_id)
        {
            if(_consumer.get()==nullptr)
            {
                DLOG("消息确认时,找不到消费者信息");
                return;
            }
            std::string rid=UUIDHelper::uuid();
            basicAckRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_message_id(msg_id);
            _codec->send(_conn,req);
            waitResponse(rid);
            return;
        }

        /*取消订阅*/
        void basicCancel()
        {
            if(_consumer.get()==nullptr) return;
            std::string rid=UUIDHelper::uuid();
            basicCancelRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->qname);
            req.set_consumer_tag(_consumer->tag);
            _codec->send(_conn,req);
            waitResponse(rid);
            _consumer.reset();
            return;
        }

        /*订阅消息,只有消费者才订阅消息*/
        bool basicConsume(const std::string& consumer_tag,
                        const std::string& queue_name,
                        bool auto_ack,
                        const ConsumerCallback& cb)
        {
            if(_consumer.get()!=nullptr)
            {
                DLOG("当前信道已经订阅其他队列消息");
                return false;
            }
            std::string rid=UUIDHelper::uuid();
            basicConsumeRequest req;
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(queue_name);
            req.set_auto_ack(auto_ack);
            _codec->send(_conn,req);
            basicCommonResponsePtr resp=waitResponse(rid);
            if(resp->ok()==false)
            {
                DLOG("添加订阅失败");
                return false;
            }
            _consumer=std::make_shared<Consumer>(consumer_tag,queue_name,auto_ack,cb);
            return true;
        }

        /*连接收到基础响应后,向hash中添加响应*/
        void putBasicResponse(const basicCommonResponsePtr& resp)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _basic_resp.insert(std::make_pair(resp->rid(),resp));
            _cond.notify_all();
        }

        /*连接收到消息推送后,需要通过信道找到对应的消费者对象,通过回调函数进行消息处理*/
        void consume(const basicConsumeResponsePtr& resp)
        {
            if(_consumer.get()==nullptr)
            {
                DLOG("消息处理时,未找到订阅者信息");
                return;
            }
            if(_consumer->tag!=resp->consumer_tag())
            {
                DLOG("收到推送消息中的消费者标识,与当前消费者标识不一致");
                return;
            }
            _consumer->callback(resp->consumer_tag(),resp->mutable_properties(),resp->body());
        }
        ~Channel()
        {
            if(_consumer.get()!=nullptr) basicCancel();
        }
    
    private:
        /*同步等待*/
        basicCommonResponsePtr waitResponse(const std::string& rid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock,[&rid,this](){
                return _basic_resp.find(rid)!=_basic_resp.end();
            });
            basicCommonResponsePtr basic_resp=_basic_resp[rid];
            _basic_resp.erase(rid);
            return basic_resp;
        }
    private:
        std::string _cid;//信道ID
        muduo::net::TcpConnectionPtr _conn;//connection链接句柄
        ProtobufCodecPtr _codec;//protobuf协议处理句柄
        Consumer::ptr _consumer;//信道关联的消费者
        std::mutex _mutex;//互斥锁
        std::condition_variable _cond;//条件变量
        std::unordered_map<std::string,basicCommonResponsePtr> _basic_resp;//信道ID与通用响应句柄哈希
    };

    class ChannelManager
    {
    public:
        using ptr=std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        /*创建信道*/
        Channel::ptr create(const muduo::net::TcpConnectionPtr& conn,
                            const ProtobufCodecPtr& codec)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto channel=std::make_shared<Channel>(conn,codec);
            _channels.insert(std::make_pair(channel->cid(),channel));
            return channel;
        }

        /*移除信道*/
        void remove(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }

        /*根据信道ID获取信道句柄*/
        Channel::ptr getChannel(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto iter=_channels.find(cid);
            if(iter==_channels.end()) return Channel::ptr();
            return iter->second;
        }
    
    private:
        std::mutex _mutex;//互斥锁
        std::unordered_map<std::string,Channel::ptr> _channels;//信道哈希
    };
}

#endif